#include "libcomm.h"
#if (COMM_FEATURE_LIBEVENT==1)

typedef std::map<std::string,IBind*> BINDMAP;
#define BINDMAPINST	((BINDMAP*)m_map)

BindMap::BindMap():m_map(NULL)
{
	m_map = new BINDMAP();
	MYLIB_ASSERT_NOTNULL(m_map);
}

BindMap::~BindMap()
{ 
	clear(); 
	if(NULL != m_map)
	{
		delete BINDMAPINST;
		m_map = NULL;
	}
}

IBind* BindMap::get(const char* classname)
{ 
	ThreadLockGuard guard(&m_lock);
	BINDMAP::iterator itor = BINDMAPINST->find(classname);
	if(itor == BINDMAPINST->end())
		return NULL;
	return itor->second;
}

void BindMap::set(const char* classname,IBind* bind)
{
	ThreadLockGuard guard(&m_lock);
	(*BINDMAPINST)[classname] = bind;
}

void BindMap::clear()
{
	ThreadLockGuard guard(&m_lock);
	for (BINDMAP::iterator itor = BINDMAPINST->begin();itor != BINDMAPINST->end();itor++)
		delete itor->second;
	BINDMAPINST->clear();
}

//////////////////////////////////////////////////////////////////////////

MYLIB_TLS MessageHandler* trd_messageHandler = NULL;
MYLIB_TLS int trd_rpcError = RPC_SUCCESS;

class RPCImp
{
public:
	enum RequestType{ RPC_MIN=100, RPC_REQUEST, RPC_RESPONSE, RPC_MAX };
	class RequestItem
	{
	public:
		MessageHandler* hanlder;
		int id;
		CommType response; 
		ThreadSemaphore sem;
	};
	class RequestItemGuard
	{
	public:
		List<RequestItem*,ThreadLock,int>* m_requestList;
		RequestItem* m_item;
		RequestItemGuard(List<RequestItem*,ThreadLock,int>* requestList,RequestItem* item)
			:m_requestList(requestList),m_item(item){ m_requestList->Append(item);}
		~RequestItemGuard(){ m_requestList->Remove(m_item->id);}
	};
	class RequestListReleaser : public CommonListReleaser<RequestItem*,int>
	{
	public:
		virtual bool Compare(RequestItem*& item, const int& value){ return (item->id==value);}
	};
	List<RequestItem*,ThreadLock,int> m_requestList;
	AtomInt m_requestid;

	RPCImp():m_requestList(new RequestListReleaser()),m_requestid(1000){}
	~RPCImp(){}
	void OnClose(MessageHandler* hanlder, SocketClient::ConnectionState reason)
	{
		// let all waiting Call() exit...
		int callCount = 0;
		{
			ThreadLockGuard guard(m_requestList.Locker());
			(void*)&reason;
			foreach(RequestItem*,item,m_requestList.list())
			{
				if(item->hanlder == hanlder)
				{
					callCount++;
					item->sem.SetSignal();
				}
			}
		}
		//if(callCount > 0)
		//	Thread::Sleep(100);
	}

	// return true mean the message had handled, would not be route to other filters. 
	bool OnMessage(MessageHandler* hanlder, const char* buffer, int size)
	{
		MYLIB_ASSERT_NOTNULL(hanlder);
		trd_messageHandler = hanlder;
		Arguments args = Arguments::Deserialize(buffer,size);
		if(args.count()<3 || args.count()>IBind::MAX_ARG_COUNT)
		{
			LOGWARN("RPC-request format error!");
			return false;
		}

		short requesttype = args[0];
		if(requesttype<=RPC_MIN || requesttype>=RPC_MAX)
		{
			LOGWARN("RPC-request type error!");
			return false;
		}

		int requestid = args[1];
		switch (requesttype)
		{
		case RPC_REQUEST:
			{
				// invoking registered RPC methods.
				//
				String classname = args[2];
				CommType result = CommType::Error;
				do 
				{
					IBind* bind = BindMap::Instance()->get(classname);
					if(NULL == bind)
					{
						LOGWARN("RPC(%d) error: un-registered class name \"%s\".",requestid,(char*)classname);
						break;
					}

					//LOGNOTICE("RPC(%d) call \"%s\"...",requestid,(char*)classname);
					args.remove(0); // remove request type.
					args.remove(0); // remove request id.
					args.remove(0); // remove method name.
					result = bind->Call(args);
					String value = String::Empty;
					{
						if(CommType::CT_BUFFER != result.Type() && CommType::CT_STRING != result.Type())
							value = result.ToString();
					}
					//LOGNOTICE("RPC(%d) call \"%s\" result: (%s) %s.",requestid,(char*)classname,result.TypeString(),(char*)value);

				} while (false);
				
				// feedback.
				//
				args.clear();
				requesttype = RPC_RESPONSE;
				args << requesttype;
				args << requestid;
				args << result;
				Buffer feedback = args.Serialize();
				if(!hanlder->Write(feedback.get(),feedback.size()))
				{
					LOGWARN("RPC(%d) call \"%s\" error: send feedback failure.",requestid,(char*)classname);
				}
			}
			break;
		case RPC_RESPONSE:
			{
				ThreadLockGuard guard(m_requestList.Locker());
				RequestItem* item = NULL;
				if(m_requestList.Fetch(item,requestid))
				{
					item->response = args[2]; 
					item->sem.SetSignal();
				}
			}
			break;
		default:
			return false;
		}

		return true; 
	}

	bool Call(MessageHandler* hanlder, const char* classname, const Arguments& args, CommType* response, unsigned int timeoutMs)
	{
		// RPC Request format:
		// short: request type.
		// int: request id.
		// string: method class name.
		// arguments.
		//
		MYLIB_ASSERT_NOTNULL(hanlder);
		SocketClient::Guard guard(hanlder->Peer()); // add by wanghh 2021.7.30
		short requesttype = RPC_REQUEST;
		int requestid = m_requestid++;
		Arguments args_all = args;
		args_all.insert(0,classname);
		args_all.insert(0,requestid);
		args_all.insert(0,requesttype);
		MYLIB_ASSERT(args_all.count()<=IBind::MAX_ARG_COUNT);
		Buffer buffer = args_all.Serialize();

		RequestItem* item = new RequestItem;
		MYLIB_ASSERT_NOTNULL(item);
		item->hanlder = hanlder;
		item->id = requestid;
		item->response = CommType::Error;
		{
			// auto remove request item.
			RequestItemGuard requestGuard(&m_requestList,item);

			// send message.
			if(!hanlder->Write(buffer.get(),buffer.size()))
			{
				LOGWARN("RPC-request(%d, %s) send failure.",item->id,classname);
				trd_rpcError = RPC_SEND_FAILURE;
				return false;
			}

			// wait response.
			if(NULL != response)
			{
				if(!item->sem.WaitSignal(timeoutMs))
				{	
					LOGWARN("RPC-request(%d, %s) waiting timeout.",item->id,classname);
					trd_rpcError = RPC_WAIT_TIMEOUT;
					return false;
				}
				if(item->response.IsError())
				{
					LOGWARN("RPC-request(%d, %s) return system error.",item->id,classname);
					trd_rpcError = RPC_SYSTEM_ERROR;
					return false;
				}
				*response = item->response;
			}
		}
		trd_rpcError = RPC_SUCCESS;
		return true;
	}
};

//////////////////////////////////////////////////////////////////////////

#define RPCIMP ((RPCImp*)m_inst_opaque)

RPCFilter::RPCFilter():m_inst_opaque(NULL)
{
	m_inst_opaque = new RPCImp();
	MYLIB_ASSERT_NOTNULL(m_inst_opaque);
}

RPCFilter::~RPCFilter()
{
	if(NULL != m_inst_opaque)
	{
		delete RPCIMP;
		m_inst_opaque = NULL;
	}
}

bool RPCFilter::Call(MessageHandler* hanlder, const char* classname, const Arguments& args, CommType* response, unsigned int timeoutMs)
{
	return RPCIMP->Call(hanlder,classname,args,response,timeoutMs);
}

bool RPCFilter::OnMessage(MessageHandler* hanlder, const char* buffer, int size)
{
	return RPCIMP->OnMessage(hanlder,buffer,size);
}

void RPCFilter::OnClose(MessageHandler* hanlder, SocketClient::ConnectionState reason)
{
	RPCIMP->OnClose(hanlder,reason);
}

//////////////////////////////////////////////////////////////////////////

class MessageFilterFactory : public MessageHandlerFactory
{
public:
	MessageFilterFactory() : m_filterList(new CustomListReleaser<MessageFilter*>()) { m_filterList.Append(new RPCFilter()); }
	virtual MessageParser* CreateParser() { return new StreamMessageParser(); }
	virtual MessageHandler* CreateHandler() { return new MessageFilterManager(&m_filterList); }

	List<MessageFilter*>& FilterList() { return m_filterList; }
private:
	List<MessageFilter*> m_filterList;
};

RPC::RPC(const SocketAddress& address, bool asClient, int dispatchThreadCount, int dispatchIntervalMs, const char* crtfile, const char* keyfile)
	:MessageReactor(address, new MessageFilterFactory(), asClient, dispatchThreadCount, dispatchIntervalMs, crtfile, keyfile)
{
}

bool RPC::Call(const char* classname, const Arguments& args, CommType* response, unsigned int timeoutMs, const char* address)
{
	MessageFilter* rpcFilter = NULL;
	this->FilterList().Peek(rpcFilter);
	MYLIB_ASSERT_NOTNULL(rpcFilter);
	MessageHandler* handler = this->Messager(address);
	if(NULL == handler)
	{
		LOGWARN("%s failure: no connections to do.",__FUNCTION__);
		trd_rpcError = RPC_CONNECT_FAILURE;
		return false;
	}
	return ((RPCFilter*)rpcFilter)->Call(handler,classname,args,response,timeoutMs);
}

List<MessageFilter*>& RPC::FilterList()
{
	MessageFilterFactory* factory = (MessageFilterFactory*)m_handlerFactory;
	return factory->FilterList();
}

MessageHandler* RPC::CurrentMessager()
{
	return trd_messageHandler;
}

RPCError RPC::CurrentError()
{
	return (RPCError)trd_rpcError;
}


#endif